package com.sunzm.flink.callrecordconnect

import java.util.Properties

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.slf4j.{Logger, LoggerFactory}

object CallRecordConnectJob {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    logger.info("开始...")

    val properties: Properties = new Properties
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.156.210.70:9093")
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink-test")
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    val callCenterMessage: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("call_center_message",
      new SimpleStringSchema, properties))
      .filter(line=>{
        var res = false
        try {
          val jSONObject: JSONObject = JSON.parseObject(line)
          val callId: String = jSONObject.getString("callid")
          if(StringUtils.isNotBlank(callId)){
            res = true
          }
        } catch {
          case e: Throwable => {
            logger.info(s"JSON解析异常:${line}", e)
          }
        }

        res
      })
    val callLogRecord: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("call_log_record",
      new SimpleStringSchema, properties)).filter(line=>{
      var res = false
      try {
        val jSONObject: JSONObject = JSON.parseObject(line)
        val callId: String = jSONObject.getString("parentCallRecordId")
        if(StringUtils.isNotBlank(callId)){
          res = true
        }
      } catch {
        case e: Throwable => {
          logger.info(s"JSON解析异常:${line}", e)
        }
      }

      res
    })

    val callCenterMessageKeyedDS: KeyedStream[String, String] = callCenterMessage.keyBy(line => {
      val callCenterMessageObject: JSONObject = JSON.parseObject(line)
      callCenterMessageObject.getString("callid")
    })
    val callLogRecordKeyedDS: KeyedStream[String, String] = callLogRecord.keyBy(line => {
      val callLogRecordObject: JSONObject = JSON.parseObject(line)
      callLogRecordObject.getString("parentCallRecordId")
    })

    val connectedStreams: ConnectedStreams[String, String] = callLogRecordKeyedDS.connect(callCenterMessageKeyedDS)

    //使用map方法
    /*connectedStreams.map(new CoMapFunction[String, String, String]{
      override def map1(input: String) = {

        input
      }

      override def map2(input: String) = {

        "NULL"
      }
    }).filter(_ != "NULL")*/

    //使用flatMap方法，可以没有输出
    val resultDS: DataStream[String] = connectedStreams.flatMap(new CoFlatMapFunction[String, String, String] {
      private var callRecord = ""
      private var callMessage = ""
      override def flatMap1(input: String, collector: Collector[String]) = {
        callRecord = input

        if(StringUtils.isNotBlank(callMessage)){
          val callMessageJSON: JSONObject = JSON.parseObject(callMessage)
          val callRecordJSON: JSONObject = JSON.parseObject(callRecord)

          getCallRecord(callMessageJSON, callRecordJSON)

          collector.collect(callRecordJSON.toJSONString)
        }
      }

      override def flatMap2(input: String, collector: Collector[String]) = {
        callMessage = input
        if(StringUtils.isNotBlank(callRecord)){
          val callMessageJSON: JSONObject = JSON.parseObject(callMessage)
          val callRecordJSON: JSONObject = JSON.parseObject(callRecord)

          getCallRecord(callMessageJSON, callRecordJSON)

          collector.collect(callRecordJSON.toJSONString)
        }
      }
    })

    resultDS.print()

    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  def getCallRecord(callMessageJSON: JSONObject, callRecordJSON: JSONObject) = {
    val satisfyLevel: Int = callMessageJSON.getIntValue(" satisfyLevel")
    val satisfyKey: String = callMessageJSON.getString("satisfyKey")
    val satisfyNumber: Int = callMessageJSON.getIntValue("satisfyNumber")
    val satisfyStr: String = callMessageJSON.getString("satisfyStr")

    callMessageJSON.clear()

    callRecordJSON.put("satisfyLevel", satisfyLevel)
    callRecordJSON.put("satisfyKey", satisfyKey)
    callRecordJSON.put("satisfyNumber", satisfyNumber)
    callRecordJSON.put("satisfyStr", satisfyStr)
    callRecordJSON.put("msgType", "completeMsg")
  }
}
